Patch/abhi tablename : Adding import query type property and modify tableName property for redshift and postgres plugin #607
Conversation
amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftSchemaReader.java
Outdated
Show resolved
Hide resolved
| /** | ||
| * Override: Fetches schema fields for a specific table using database metadata. | ||
| */ | ||
| @Override |
There was a problem hiding this comment.
It is there in parent class, you can use that one, instead you can call getSchema method of this class inside
There was a problem hiding this comment.
no update on this, also javadoc not updated
| public void configurePipeline(PipelineConfigurer pipelineConfigurer) { | ||
| FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); | ||
| if (!sourceConfig.containsMacro("tableName") && !sourceConfig.containsMacro("importQuery")) { | ||
| if ((sourceConfig.getTableName() == null || sourceConfig.getTableName().isEmpty()) |
There was a problem hiding this comment.
Use Strings method for null empty check
amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftSource.java
Outdated
Show resolved
Hide resolved
| @@ -88,6 +107,11 @@ protected LineageRecorder getLineageRecorder(BatchSourceContext context) { | |||
| return new LineageRecorder(context, assetBuilder.build()); | |||
| } | |||
|
|
|||
| public DatabaseMetaData getDatabaseMetadata(Connection connection) throws SQLException { | |||
| return (DatabaseMetaData) connection.getMetaData().getColumns(null, | |||
| null, redshiftSourceConfig.getTableName(), null); | |||
There was a problem hiding this comment.
schema is not handled here, if customer is passing "schema.tablename", it will not work
There was a problem hiding this comment.
for now this method is not used so no need to handle we can remove this method .
There was a problem hiding this comment.
if not getting used then remove it
| @@ -251,6 +275,30 @@ | |||
| "name": "port" | |||
| } | |||
| ] | |||
| }, | |||
There was a problem hiding this comment.
no need to add filters as it is hidden
There was a problem hiding this comment.
keeping it so that if further we have to make changes for this we can just changed the hidden property.
but if not required , will removed..
There was a problem hiding this comment.
remove for now, will add later if required
database-commons/src/main/java/io/cdap/plugin/db/CommonSchemaReader.java
Outdated
Show resolved
Hide resolved
database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java
Outdated
Show resolved
Hide resolved
database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java
Outdated
Show resolved
Hide resolved
| if (!Strings.isNullOrEmpty(importQuery)) { | ||
| return loadSchemaFromDB(connection, importQuery); | ||
| } else { | ||
| String query = String.format("SELECT * FROM %s LIMIT 1", tableName); |
There was a problem hiding this comment.
we have fetch schema from database metadata in case of tableName, this should not behave like this
| @@ -73,7 +72,7 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti | |||
| return Schema.of(Schema.Type.STRING); | |||
| } | |||
| } | |||
|
|
|||
There was a problem hiding this comment.
not removed properly, this class has no changes it should not be there in commit files
There was a problem hiding this comment.
removed now and shown in commit file
postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresSource.java
Outdated
Show resolved
Hide resolved
| public void configurePipeline(PipelineConfigurer pipelineConfigurer) { | ||
| FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); | ||
| if (!sourceConfig.containsMacro("tableName") && !sourceConfig.containsMacro("importQuery")) { | ||
| if ((Strings.isNullOrEmpty(sourceConfig.getTableName())) |
There was a problem hiding this comment.
merge these two conditions
| @@ -61,4 +62,40 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti | |||
| public boolean shouldIgnoreColumn(ResultSetMetaData metadata, int index) throws SQLException { | |||
| return false; | |||
| } | |||
|
|
|||
|
|
||
|
|
||
| @Override | ||
| public List<Schema.Field> getSchemaFields(Connection connection, String tableName) throws SQLException { |
There was a problem hiding this comment.
add javadoc whenever adding a public method
| if (!containsMacro(IMPORT_QUERY) && Strings.isNullOrEmpty(importQuery)) { | ||
| collector.addFailure("Import Query is empty.", "Specify the Import Query.") | ||
| .withConfigProperty(IMPORT_QUERY); | ||
| if (!containsMacro(TABLE_NAME) && !containsMacro(IMPORT_QUERY)) { |
There was a problem hiding this comment.
merge if conditions
| "Import Query %s must contain the string '$CONDITIONS'. if Number of Splits is not set to 1.", importQuery), | ||
| "Include '$CONDITIONS' in the Import Query") | ||
| .withConfigProperty(IMPORT_QUERY); | ||
| if (!Strings.isNullOrEmpty(importQuery)) { |
There was a problem hiding this comment.
merge if conditions
| @@ -178,7 +202,7 @@ public void validateSchema(Schema actualSchema, FailureCollector collector) { | |||
| Schema expectedFieldSchema = field.getSchema().isNullable() ? | |||
| field.getSchema().getNonNullable() : field.getSchema(); | |||
|
|
|||
| validateField(collector, field, actualFieldSchema, expectedFieldSchema); | |||
| validateField(collector, field, actualFieldSchema, expectedFieldSchema); | |||
| { | ||
| "name": "ImportQuery", | ||
| "condition": { | ||
| "expression": "importQueryType == 'importQuery'" |
There was a problem hiding this comment.
this is not done as per redshift
adding import query type property to db plugin.which will only reflect in redshift and Postgres plugin.
AbhishekKumar9984
force-pushed
the
patch/abhi-tablename
branch
from
f4d0f7c to
8826eac
Compare
Adding import query type property and modify tableName property for redshift and postgres plugin